package com.JadePenG.demo;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;

/**
 * 组装topology结构, 将代码上传到集群上运行
 */
public class WCTopologyMain {
    //本地的配置
    /*public static void main(String[] args) {
        //1.组装topology结构
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        //1.1设置spout 参数1：组件名称 参数2：spout对象
        topologyBuilder.setSpout("rfSpout", new ReadFileSpout());
        //1.2设置splitbolt   分组策略，上游该如何向下游发送数据  shuffleGrouping 随机分组
        //标明上游是谁 --> rfSpout
        topologyBuilder.setBolt("slBolt", new SplitBolt()).shuffleGrouping("rfSpout");
        //1.3设置wcBolt
        //标明上游是谁 --> slBolt
        topologyBuilder.setBolt("wcBolt", new WordCountBolt()).shuffleGrouping("slBolt");

        //2.将代码上传到集群上运行
        //运行方式有两种： 1，本地运行 IDEA LocalCluster本地集群   2.集群运行
        LocalCluster localCluster = new LocalCluster();

        //strop HashMap  配置对象, 采用默认的配置即可
        Config config = new Config();
        //提交到集群    参数1：topology名称 字符串  参数2：map  storm的集群配置对象 参数3：topology对象
        localCluster.submitTopology("wordCount", config, topologyBuilder.createTopology());

    }*/

    //增  在集群上的配置
    public static void main(String[] args) throws Exception {
        //1.组装topology结构
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        //1.1设置spout 参数1：组件名称 参数2：spout对象
        topologyBuilder.setSpout("rfspout", new ReadFileSpout(), 2);
        //1.2设置splitbolt   分组策略，上游该如何向下游发送数据
        topologyBuilder.setBolt("slbolt", new SplitBolt(), 2).shuffleGrouping("rfspout");
        //1.3设置wcBolt
        topologyBuilder.setBolt("wcbolt", new WordCountBolt(), 2).shuffleGrouping("slbolt");

        //2.将代码上传到集群上运行
        //运行方式有两种： 1，本地运行 IDEA   2.集群运行
        Config config = new Config();//配置对象，这里采用默认的即可
        config.setMaxSpoutPending(3000);//如果内存中有3000条数据失败了那么，spout就不在发送数据了
        //config.setNumAckers(1);//设定worker中ack组件的个数的，默认是1 如果开启ack  数量要>=1
        config.setNumWorkers(3);//默认worker数是1 本机运行设定worker数是没有意义的
        if (args != null && args.length > 0) {
            //集群运行
            StormSubmitter.submitTopology(args[0], config, topologyBuilder.createTopology());
        } else {
            LocalCluster localCluster = new LocalCluster();
            //参数1：topology名称 字符串  参数2：map  storm的集群配置对象 参数3：topology对象
            localCluster.submitTopology("wordcount", config, topologyBuilder.createTopology());
        }

    }
}
